package mqtt

import (
	"bytes"
	"compress/gzip"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	logger "eventalarm2taos/logger"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

type MsgData struct {
	Payload []byte
	Topic   []string
}

type MqttStru struct {
	Client       mqtt.Client
	sendTopic    string
	compress     bool
	ClientID     string
	Subscribe    []string
	MsgChannel   chan MsgData
	LastRece     time.Time
	ReConnectCnt int
}

// Init 初始化MqttStru结构体中的MQTT客户端
//
// 参数：
//   - firstTopic: 初始订阅的主题
//   - isCompress: 是否启用消息压缩
//   - user: MQTT用户名
//   - pwd: MQTT密码
//   - server: MQTT服务器地址
//   - port: MQTT服务器端口
//   - caPath: CA证书文件路径
//   - onLost: 连接丢失时的回调函数
//   - onConnect: 连接建立时的回调函数
//
// 返回值：
//   - err: 错误信息，如果初始化成功则为nil
//
// 功能：
//  1. 初始化MqttStru结构体中的相关变量
//  2. 创建一个MQTT客户端选项
//  3. 设置用户名、密码、保活时间、ClientID等选项
//  4. 读取CA证书文件并设置TLS配置（如果提供了caPath）
//  5. 设置MQTT代理的地址
//  6. 设置连接丢失和连接建立的回调函数
//  7. 创建MQTT客户端
//  8. 尝试连接MQTT代理
//  9. 返回连接结果
func (m *MqttStru) Init(firstTopic string, isCompress bool, user string, pwd string, server string, port int, caPath string,
	onLost mqtt.ConnectionLostHandler, onConnect mqtt.OnConnectHandler) (err error) {

	logger.Info(fmt.Sprintf("init mqtt client! clientID[%s] topic[%s] ip[%s] port[%d] cafile[%s]",
		m.ClientID,
		firstTopic,
		server,
		port,
		caPath))

	//保存变量
	m.sendTopic = fmt.Sprintf("%s/%s", firstTopic, m.ClientID)
	m.compress = isCompress

	// 创建MQTT客户端选项
	opts := mqtt.NewClientOptions()

	// 设置用户名和密码
	opts.SetUsername(user)
	opts.SetPassword(pwd)

	// 设置保活时间为 600 秒
	opts.SetKeepAlive(60)

	//设置mqtt的clientid
	opts.SetClientID(m.ClientID)

	// 如果存在ca读取证书
	broker := ""
	if caPath != "" {
		ca, err := os.ReadFile(caPath)
		if err != nil {
			return fmt.Errorf("init:ca-> %w", err)
		}
		caCertPool := x509.NewCertPool()
		ok := caCertPool.AppendCertsFromPEM(ca)
		if !ok {
			return fmt.Errorf("init:x509-> ca fail")
		}

		tlsConfig := &tls.Config{RootCAs: caCertPool}
		opts.SetTLSConfig(tlsConfig)
		broker += "ssl://"
	}

	// 设置MQTT代理的地址
	broker = broker + server + ":" + strconv.Itoa(port)
	opts.AddBroker(broker)

	opts.OnConnectionLost = onLost
	opts.OnConnect = onConnect

	// 创建MQTT客户端
	m.Client = mqtt.NewClient(opts)

	// 连接MQTT代理
	token := m.Client.Connect()
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("init:connect-> %w", token.Error())
	}
	return nil
}

// Send 函数，发送数据。
//
// 参数:
//
//	data - 要发送的数据，二进制格式
//	varargs - 发送主题，如果不输入，默认使用初始化时的sendTopic
//
// 返回值:
//
//	err - 发生的异常
//
// 示例:
func (m *MqttStru) Send(data []byte, varargs ...string) (err error) {

	if data == nil {
		return nil
	}

	payload := data
	if m.compress {
		// Use Golang's built-in gzip package to compress the payload
		msg := make([]byte, 0)
		var err error
		buf := bytes.NewBuffer(msg)
		gz := gzip.NewWriter(buf)
		_, err = gz.Write(data)
		if err != nil {
			return fmt.Errorf("send:write-> %w", err)
		}
		err = gz.Flush()
		if err != nil {
			return fmt.Errorf("send:flush-> %w", err)
		}
		err = gz.Close()
		if err != nil {
			return fmt.Errorf("send:close-> %w", err)
		}
		payload = buf.Bytes()
	}

	//发布消息
	var topic string
	if len(varargs) == 0 {
		topic = m.sendTopic
	} else {
		topic = varargs[0]
	}

	token := m.Client.Publish(topic, 0, false, payload)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("send:publish-> %w", token.Error())
	}
	return nil
}

// Close 关闭MqttStru的连接
// 如果MqttStru中的Client连接是打开的，则调用Client的Disconnect方法断开连接
// 参数：
//
//	无
//
// 返回值：
//
//	err error：如果关闭连接时发生错误，则返回非零的错误码；否则返回nil
func (m *MqttStru) Close() (err error) {

	// 断开连接
	if m.Client.IsConnectionOpen() {
		m.Client.Disconnect(200)
	}
	return nil
}

// Recv 方法从MQTT客户端接收消息
//
// 参数：
//
//	m *MqttStru：MQTT结构的指针，包含了MQTT客户端和其他相关配置
//
// 返回值：
//
//	err error：如果接收过程中发生错误，返回错误；否则返回nil
func (m *MqttStru) Recv() (err error) {
	// 订阅主题
	topics := make(map[string]byte)
	for _, item := range m.Subscribe {
		topics[item] = 0
	}

	token := m.Client.SubscribeMultiple(topics,
		func(client mqtt.Client, msg mqtt.Message) {
			//r1是有缓冲通道，此处不会阻塞
			m.MsgChannel <- MsgData{
				Payload: msg.Payload(),
				Topic:   strings.Split(msg.Topic(), "/"),
			}
		},
	)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("recv-> %w", token.Error())
	}
	return nil
}
